package scheduler

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/goccy/go-json"
	"github.com/mitchellh/mapstructure"

	"gitee.com/wujianqiang/golibs/pkg/factory"
	"gitee.com/wujianqiang/golibs/pkg/logging"
	"github.com/nats-io/nats.go"
)

// 负责管理任务调度
// SchedulerPolicy 中的name 与  Event中的name 一致，Scheduler将执行Scheduler策略，决定当前event是否结束，
// 否则所有event 执行一次就结束

// webhooks 的key 是 event name,event 调用一次触发webhooks执行
type Scheduler struct {
	Tasks       map[string]*PipeLine
	WebHooks    map[string]*WebHook
	NatClient   *factory.NatClient
	SubmitQueue string
	SubmitTopic string
	WorkerTopic string
	sync.RWMutex

	ActionDelQueue    chan DelTask
	ActionUpdateQueue chan SubTask
	ActionSubQueue    chan SubTask

	SchedulerPolicy map[string]SchedulerPolicy
}

func (s *Scheduler) AddWebHook(event EventType, wh *WebHook) error {
	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()
	s.WebHooks[string(event)] = wh
	return nil
}

func (s *Scheduler) AddSchedulerPolicy(event EventType, sp SchedulerPolicy) error {
	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()
	s.SchedulerPolicy[string(event)] = sp
	return nil
}

func (s *Scheduler) AddTask(task *PipeLine) error {
	if gettask, er := s.PopPipeline(task.ID); er == nil {
		go func(pipeline *PipeLine) {
			s.DelTask(pipeline, nil)
		}(gettask)
	}
	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()
	s.Tasks[task.ID] = task
	return nil
}

func (s *Scheduler) DelTask(pipeline *PipeLine, cancel *CancelTask) error {

	//当前任务停止
	if pipeline.GetSchedulerRun() {
		pipeline.Cancel()
	}
	eventargs := pipeline.EventsResult

	//按照要求cancel event，需要从当前的event  cancel-->设置event
	if cancel != nil {
		endindex, er := pipeline.GetIndexEvent(cancel.EndEvent)
		if er != nil {
			logging.Logger.Error("GetIndexEvent:", er, "EVENT:", cancel.EndEvent)
			return er
		}
		for {
			if pipeline.Current >= endindex {
				event, args := pipeline.GetCurrentEvent()
				if runargs, ok := eventargs[event]; ok {
					args = runargs.(map[string]interface{})
				}
				s.ExecCancelEvent(context.Background(), event, args)
				if er := pipeline.PreEvent(); er != nil {
					logging.Logger.Error("PreEvent:", er)
					logging.Logger.Info("pipeline:", pipeline)
					logging.Logger.Info("cancelTask:", cancel)
					break
				}
			} else {
				break
			}
		}
	} else {
		//添加时，pipeline已存在，需要删除
		for {
			event, _ := pipeline.GetCurrentEvent()
			if args, ok := eventargs[event]; ok {
				s.ExecCancelEvent(context.Background(), event, args.(map[string]interface{}))
			} else {
				s.ExecCancelEvent(context.Background(), event, pipeline.Args)
			}
			er := pipeline.PreEvent()
			if er != nil {
				break
			}
		}

	}

	return nil
}

func (s *Scheduler) DelAndExecTask(del DelTask) error {
	if pipeline, er := s.PopPipeline(del.CancelTask.JobId); er == nil {
		s.DelTask(pipeline, &del.CancelTask)
	}
	success := map[string]interface{}{
		"code": 0,
	}
	s.NatClient.ReplyData(success, del.Msg.Reply)
	return nil

}

func NewScheduler(config SchedulerConfig) (*Scheduler, error) {
	natclient, er := config.NatConf.Connected()
	if er != nil {
		return nil, er
	}
	s := &Scheduler{
		Tasks:             make(map[string]*PipeLine),
		WebHooks:          make(map[string]*WebHook),
		SchedulerPolicy:   make(map[string]SchedulerPolicy),
		NatClient:         natclient,
		SubmitQueue:       config.SubmitQueue,
		SubmitTopic:       config.SubmitTopic,
		WorkerTopic:       config.WorkerTopic,
		ActionDelQueue:    make(chan DelTask, 1024),
		ActionUpdateQueue: make(chan SubTask, 1024),
		ActionSubQueue:    make(chan SubTask, 1024),
	}
	return s, nil
}

func (s *Scheduler) Run() {
	//{action: sub   data: subTask}
	s.NatClient.Conn.QueueSubscribe(s.SubmitTopic, s.SubmitQueue, func(msg *nats.Msg) {
		var rescv map[string]interface{}
		err := json.Unmarshal(msg.Data, &rescv)
		logging.Logger.Info("scheduler get  job:", rescv)
		if err != nil {
			resmap := map[string]interface{}{
				"code":  1,
				"ermsg": err.Error(),
			}
			s.NatClient.ReplyData(resmap, msg.Reply)
			return
		}
		switch rescv["action"].(string) {
		case "sub":
			var subTask SubTask
			er := mapstructure.Decode(rescv["data"], &subTask)
			if er != nil {
				resmap := map[string]interface{}{
					"code":  1,
					"ermsg": er.Error(),
				}
				s.NatClient.ReplyData(resmap, msg.Reply)
				return
			}
			logging.Logger.Info("get  job:", subTask)
			s.ActionSubQueue <- subTask
		case "update":
			var subTask SubTask
			er := mapstructure.Decode(rescv["data"].(map[string]interface{}), &subTask)
			if er != nil {
				resmap := map[string]interface{}{
					"code":  1,
					"ermsg": er.Error(),
				}
				s.NatClient.ReplyData(resmap, msg.Reply)
				return
			}
			s.ActionUpdateQueue <- subTask
		case "del":
			var delid CancelTask
			er := mapstructure.Decode(rescv["data"].(map[string]interface{}), &delid)
			if er != nil {
				resmap := map[string]interface{}{
					"code":  1,
					"ermsg": er.Error(),
				}
				s.NatClient.ReplyData(resmap, msg.Reply)
				return
			}
			s.ActionDelQueue <- DelTask{CancelTask: delid, Msg: msg}
		}

	})

	for {
		select {
		case delid := <-s.ActionDelQueue:
			// 删除任务
			logging.Logger.Info("scheduler del:", delid)
			go func() {
				s.DelAndExecTask(delid)

			}()

		case updatejob := <-s.ActionUpdateQueue:
			// 更新任务
			logging.Logger.Info("scheduler updatejob:", updatejob)
			go func() {
				s.UpdateTaskToPipeline(&updatejob)
			}()

		case subjob := <-s.ActionSubQueue:
			// 提交任务
			logging.Logger.Info("scheduler subjob:", subjob)
			pipeline := subjob.ToPipeline()
			logging.Logger.Info("scheduler subtask to pipeline:", pipeline)
			s.AddTask(pipeline)

		default:
			time.Sleep(time.Second * 1)
		}
		if len(s.Tasks) == 0 {
			time.Sleep(time.Second * 1)
		}
		logging.Logger.Infof("scheduler tasks:%+v", s.Tasks)
		for key, pipeline := range s.Tasks {
			if pipeline.GetSchedulerRun() {
				continue
			}
			go func(pipeline *PipeLine, key string) {
				pipeline.SetSchedulerRun(true)
				defer pipeline.SetSchedulerRun(false)
				logging.Logger.Info("scheduler run:", pipeline)
				event, args := pipeline.GetCurrentEvent()
				eventres, er := s.ExecActionEvent(pipeline.Context, event, args)
				if er != nil {
					return
				}
				pipeline.EventsResult[event] = eventres
				isComplete, er := s.ExecPolicy(event, eventres)
				if er == nil && isComplete {
					nexter := s.TasksPipelineNext(key)
					if nexter != nil {
						//任务结束了,del 当前pipeline
						s.PopPipeline(key)
					}

				}

			}(pipeline, key)

		}
	}

}

// pop 出Pipeline
func (s *Scheduler) PopPipeline(piplineId string) (*PipeLine, error) {
	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()
	if pipeline, ok := s.Tasks[piplineId]; !ok {
		return nil, errors.New("can't find pipeline")

	} else {
		delete(s.Tasks, piplineId)
		return pipeline, nil
	}
}

func (s *Scheduler) ExecActionEvent(ctx context.Context, event string, args map[string]interface{}) (map[string]interface{}, error) {
	work := WokerEvent{
		Event:   event,
		Payload: args,
		Action:  "sub",
	}
	payload, _ := json.Marshal(work)
	webhook := *s.GetWebHook(event)
	if webhook != nil && webhook.CheckWhen(WebHookWhenBefore, work) {
		webhook.Exec(WebHookEventData{
			When:   "before",
			Event:  event,
			Code:   "0",
			Action: "sub",
			Data:   args,
		})

	}
	msg, er := s.NatClient.Conn.RequestWithContext(ctx, s.WorkerTopic, payload)
	if er != nil {
		return nil, er
	}
	var res map[string]interface{}
	er = json.Unmarshal(msg.Data, &res)
	if er != nil {
		return nil, er
	}
	if webhook != nil && webhook.CheckWhen(WebHookWhenAfter, res) {
		webhook.Exec(WebHookEventData{
			When:   "after",
			Event:  event,
			Code:   "0",
			Action: "sub",
			Data:   res,
		})
	}
	return res, nil
}
func (s *Scheduler) ExecCancelEvent(ctx context.Context, event string, args map[string]interface{}) (interface{}, error) {
	work := WokerEvent{
		Event:   event,
		Payload: args,
		Action:  "cancel",
	}
	payload, _ := json.Marshal(work)
	webhook := *s.GetWebHook(event)
	if webhook != nil && webhook.CheckWhen(WebHookWhenBefore, work) {
		webhook.Exec(WebHookEventData{
			When:   "before",
			Event:  event,
			Code:   "0",
			Action: "cancel",
			Data:   args,
		})
	}
	msg, er := s.NatClient.Conn.RequestWithContext(ctx, s.WorkerTopic, payload)
	if er != nil {
		return nil, er
	}
	var res map[string]interface{}
	if msg.Data != nil {
		er = json.Unmarshal(msg.Data, &res)
		if er != nil {
			return nil, er
		}
	} else {
		res = map[string]interface{}{
			"code": 0,
		}
	}
	if webhook != nil && webhook.CheckWhen(WebHookWhenAfter, res) {
		webhook.Exec(
			WebHookEventData{
				When:   "after",
				Event:  event,
				Code:   "0",
				Action: "cancel",
				Data:   res,
			})
	}

	return res, nil

}

// bool 是否完成
func (s *Scheduler) ExecPolicy(event string, eventres interface{}) (bool, error) {
	if scheduler, ok := s.SchedulerPolicy[event]; ok {
		return scheduler.SchedulerEnd(eventres)
	}
	return true, nil
}

func (s *Scheduler) GetWebHook(event string) *WebHook {
	if webhook, ok := s.WebHooks[event]; ok {
		return webhook
	}
	return nil
}
func (s *Scheduler) UpdateTaskToPipeline(ut *SubTask) error {
	pipeline, er := s.PopPipeline(ut.JobId)
	if er != nil {
		return er
	}
	if pipeline.GetSchedulerRun() {
		pipeline.Cancel()
	}
	newpipeline := ut.ToPipeline()
	event, _ := newpipeline.GetCurrentEvent()
	//获取当前执行的event
	index, er := pipeline.GetIndexEvent(event)
	if er != nil {
		return er
	}
	if pipeline.Current >= index {
		//从当前Current 到index 的event都需要取消
		for i := pipeline.Current; i >= index; i-- {
			event := pipeline.Events[i]
			s.ExecCancelEvent(context.Background(), event, pipeline.EventsResult[event].(map[string]interface{}))
		}
	}

	return s.AddTask(newpipeline)
}

func (s *Scheduler) TasksPipelineNext(key string) error {
	s.RWMutex.Lock()
	defer s.RWMutex.Unlock()
	return s.Tasks[key].NextEvent()
}
